package com.shujia.mon

import java.util
import java.util.Properties

import com.shujia.bean.Car
import com.shujia.tf.CarMapper
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import redis.clients.jedis.Jedis

/**
  * 缉查布控
  *
  */
object TourismOfficialsMonitor {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(10000)

    // 高级选项：

    // 设置模式为精确一次 (这是默认值)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    // 开启在 job 中止后仍然保留的 externalized checkpoints
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    // 允许在有更近 savepoint 时回退到 checkpoint
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)

    /*    val stateBackend: StateBackend = new RocksDBStateBackend("hdfs://master:9000/car/checkpoint", true)
        env.setStateBackend(stateBackend)*/


    //读取卡口过车数据
    val properties = new Properties
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "asdasdasd")

    //创建消费者

    val flinkKafkaConsumer = new FlinkKafkaConsumer[String](
      "cars",
      new SimpleStringSchema,
      properties)

    flinkKafkaConsumer.setStartFromEarliest()


    val carsDS: DataStream[String] = env.addSource(flinkKafkaConsumer)


    val carDS: DataStream[Car] = carsDS.map(new CarMapper)


    //缉查布控
    val filterDS: DataStream[Car] = carDS.filter(new RichFilterFunction[Car] {

      var jedis: Jedis = _

      override def open(parameters: Configuration): Unit = {
        jedis = new Jedis("master", 6379)
      }

      override def close(): Unit = {
        jedis.close()
      }

      override def filter(car: Car): Boolean = {
        val cars: util.Set[String] = jedis.smembers("mon_list")

        //判断车辆是否是布控列表中的车辆
        cars.contains(car.getCar)
      }
    })


    //将过滤到的数据保存到redis中
    filterDS.addSink(new RichSinkFunction[Car] {
      var jedis: Jedis = _

      override def open(parameters: Configuration): Unit = {
        jedis = new Jedis("master", 6379)
      }

      override def close(): Unit = {
        jedis.close()
      }


      override def invoke(line: Car, context: SinkFunction.Context[_]): Unit = {

        val key: String = "TourismOfficialsMonitor:" + line.getCar + ":" + line.getTime

        val map = new util.HashMap[String, String]()
        map.put("city_code", line.getCity_code)
        map.put("county_code", line.getCounty_code)
        map.put("card", line.getCard.toString)
        map.put("camera_id", line.getCamera_id.toString)
        map.put("road_id", line.getRoad_id.toString)
        map.put("speed", line.getSpeed.toString)

        //以map结构保存数据
        jedis.hmset(key, map)

      }
    })


    env.execute()


  }

}
